package beautiful.butterfly.drds.data_exchange.taskgroup;


import beautiful.butterfly.drds.data_exchange.constant.Constants;
import beautiful.butterfly.drds.data_exchange.constant.PluginType;
import beautiful.butterfly.drds.data_exchange.enums.State;
import beautiful.butterfly.drds.data_exchange.exception.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.plugin.TaskPluginCollector;
import beautiful.butterfly.drds.data_exchange.statistics.communication.Communication;
import beautiful.butterfly.drds.data_exchange.statistics.plugin.task.AbstractTaskPluginCollector;
import beautiful.butterfly.drds.data_exchange.taskgroup.runner.AbstractRunner;
import beautiful.butterfly.drds.data_exchange.taskgroup.runner.ReaderRunner;
import beautiful.butterfly.drds.data_exchange.taskgroup.runner.WriterRunner;
import beautiful.butterfly.drds.data_exchange.transport.exchanger.BufferedTransformerExchanger;
import beautiful.butterfly.drds.data_exchange.transport.exchanger.Produce;
import beautiful.butterfly.drds.data_exchange.transport.exchanger.QueueExchanger;
import beautiful.butterfly.drds.data_exchange.transport.record_queue.RecordQueue;
import beautiful.butterfly.drds.data_exchange.transport.transformer.TransformerExecution;
import beautiful.butterfly.drds.data_exchange.util.Classs;
import beautiful.butterfly.drds.data_exchange.util.Configuration;
import beautiful.butterfly.drds.data_exchange.util.FrameworkErrorCode;
import beautiful.butterfly.drds.data_exchange.util.TransformerUtil;
import org.apache.commons.lang3.Validate;

import java.util.List;

/**
 * TaskExecutor是一个完整task的执行器
 * 其中包括1：1的reader和writer
 */
class TaskExecutor
{
    private TaskGroupContainer taskGroupContainer;
    private Configuration taskConfig;

    private int taskId;

    private int attemptCount;

    private RecordQueue recordQueue;

    private Thread readerThread;

    private Thread writerThread;

    private ReaderRunner readerRunner;

    private WriterRunner writerRunner;

    /**
     * 该处的taskCommunication在多处用到：
     * 1. recordQueue
     * 2. readerRunner和writerRunner
     * 3. reader和writer的taskPluginCollector
     */
    private Communication taskCommunication;

    public TaskExecutor(TaskGroupContainer taskGroupContainer, Configuration taskConf, int attemptCount)
    {
        this.taskGroupContainer = taskGroupContainer;
        // 获取该taskExecutor的配置
        this.taskConfig = taskConf;
        Validate.isTrue(null != this.taskConfig.getConfigureDataMap(Constants.job_reader)
                        && null != this.taskConfig.getConfigureDataMap(Constants.job_writer),
                "[reader|writer]的插件参数不能为空!");

        // 得到taskId
        this.taskId = this.taskConfig.getInt(Constants.task_id);
        this.attemptCount = attemptCount;

        /**
         * 由taskId得到该taskExecutor的Communication
         * 要传给readerRunner和writerRunner，同时要传给channel作统计用
         */
        this.taskCommunication = taskGroupContainer.containerCommunicator
                .getCommunication(taskId);
        Validate.notNull(this.taskCommunication,
                String.format("taskId[%d]的Communication没有注册过", taskId));
        this.recordQueue = Classs.newInstance(taskGroupContainer.channelClassName, RecordQueue.class, taskGroupContainer.configuration);
        this.recordQueue.setCommunication(this.taskCommunication);

        /**
         * 获取transformer的参数
         */

        List<TransformerExecution> transformerInfoExecs = TransformerUtil.buildTransformerInfo(taskConfig);

        /**
         * 生成writerThread
         */
        writerRunner = (WriterRunner) generateRunner(PluginType.writer);
        this.writerThread = new Thread(writerRunner, String.format("%d-%d-%d-writer", taskGroupContainer.jobId, taskGroupContainer.taskGroupId, this.taskId));


        /**
         * 生成readerThread
         */
        readerRunner = (ReaderRunner) generateRunner(PluginType.reader, transformerInfoExecs);
        this.readerThread = new Thread(readerRunner, String.format("%d-%d-%d-reader", taskGroupContainer.jobId, taskGroupContainer.taskGroupId, this.taskId));

    }

    public void doStart()
    {
        this.writerThread.start();

        // reader没有起来，writer不可能结束
        if (!this.writerThread.isAlive() || this.taskCommunication.getState() == State.failed)
        {
            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR,
                    this.taskCommunication.getThrowable());
        }

        this.readerThread.start();

        // 这里reader可能很快结束
        if (!this.readerThread.isAlive() && this.taskCommunication.getState() == State.failed)
        {
            // 这里有可能出现Reader线上启动即挂情况 对于这类情况 需要立刻抛出异常
            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR,
                    this.taskCommunication.getThrowable());
        }

    }


    public AbstractRunner generateRunner(PluginType pluginType)
    {
        return generateRunner(pluginType, null);
    }

    public AbstractRunner generateRunner(PluginType pluginType, List<TransformerExecution> transformerInfoExecs)
    {
        AbstractRunner newRunner = null;
        TaskPluginCollector pluginCollector;

        switch (pluginType)
        {
            case reader:
                //newRunner = Loaders.loadPluginRunner(pluginType, this.taskConfig.getString(Constants.job_reader_name));
                newRunner = null;//
                newRunner.setJobConf(this.taskConfig.getConfigureDataMap(
                        Constants.job_reader_parameter));

                pluginCollector = Classs.newInstance(
                        taskGroupContainer.taskCollectorClass, AbstractTaskPluginCollector.class,
                        taskGroupContainer.configuration, this.taskCommunication,
                        PluginType.reader);

                Produce produce;
                if (transformerInfoExecs != null && transformerInfoExecs.size() > 0)
                {
                    produce = new BufferedTransformerExchanger(taskGroupContainer.taskGroupId, this.taskId, this.recordQueue, this.taskCommunication, pluginCollector, transformerInfoExecs);
                } else
                {
                    produce = new QueueExchanger(this.recordQueue, pluginCollector);
                }

                ((ReaderRunner) newRunner).setProduce(produce);

                /**
                 * 设置taskPlugin的collector，用来处理脏数据和job/task通信
                 */
                newRunner.setTaskPluginCollector(pluginCollector);
                break;
            case writer:
                //newRunner = Loaders.loadPluginRunner(pluginType, this.taskConfig.getString(Constants.job_writer_name));
                newRunner = null;//
                newRunner.setJobConf(this.taskConfig
                        .getConfigureDataMap(Constants.job_writer_parameter));

                pluginCollector = Classs.newInstance(
                        taskGroupContainer.taskCollectorClass, AbstractTaskPluginCollector.class,
                        taskGroupContainer.configuration, this.taskCommunication,
                        PluginType.writer);
                ((WriterRunner) newRunner).setConsume(new QueueExchanger(
                        this.recordQueue, pluginCollector));
                /**
                 * 设置taskPlugin的collector，用来处理脏数据和job/task通信
                 */
                newRunner.setTaskPluginCollector(pluginCollector);
                break;
            default:
                throw DataExchangeException.asDataExchangeException(FrameworkErrorCode.ARGUMENT_ERROR, "Cant generateRunner for:" + pluginType);
        }

        newRunner.setTaskGroupId(taskGroupContainer.taskGroupId);
        newRunner.setTaskId(this.taskId);
        newRunner.setRunnerCommunication(this.taskCommunication);

        return newRunner;
    }

    // 检查任务是否结束
    public boolean isTaskFinished()
    {
        // 如果reader 或 writer没有完成工作，那么直接返回工作没有完成
        if (readerThread.isAlive() || writerThread.isAlive())
        {
            return false;
        }

        if (taskCommunication == null || !taskCommunication.isFinished())
        {
            return false;
        }

        return true;
    }

    public int getTaskId()
    {
        return taskId;
    }

    public long getTimeStamp()
    {
        return taskCommunication.getTimestamp();
    }

    public int getAttemptCount()
    {
        return attemptCount;
    }

    public boolean supportFailOver()
    {
        return writerRunner.supportFailOver();
    }

    public void shutdown()
    {
        writerRunner.shutdown();
        readerRunner.shutdown();
        if (writerThread.isAlive())
        {
            writerThread.interrupt();
        }
        if (readerThread.isAlive())
        {
            readerThread.interrupt();
        }
    }

    public boolean isShutdown()
    {
        return !readerThread.isAlive() && !writerThread.isAlive();
    }
}
